from kafka import KafkaConsumer, TopicPartition
import json

# 配置Kafka消费者
consumer = KafkaConsumer(
    bootstrap_servers=['47.97.221.154:9092'],  # Kafka broker地址
    auto_offset_reset='latest',  # 从最新的消息开始消费
    enable_auto_commit=True,  # 自动提交偏移量
    group_id='ssjom_proj-dev-logs_collection',  # 消费者组ID
    value_deserializer=lambda v: json.loads(v.decode('utf-8'))  # 将消息反序列化为JSON格式
)

# 获取主题的分区信息
topic = 'SSJOM_PROJ-LOGS_COLLECTION'
partitions = consumer.partitions_for_topic(topic)

# 获取每个分区的最新偏移量
latest_offsets = {}
for partition in partitions:
    tp = TopicPartition(topic, partition)
    consumer.assign([tp])
    consumer.seek_to_end(tp)
    latest_offsets[tp] = consumer.position(tp)

# 从每个分区的最新偏移量开始消费，获取最近的10条消息
recent_messages = []
for tp, offset in latest_offsets.items():
    consumer.seek(tp, max(0, offset - 100))
    for message in consumer:
        recent_messages.append(message.value)
        if len(recent_messages) >= 100:
            break
    if len(recent_messages) >= 100:
        break

# 打印最近的10条消息
for i, message in enumerate(recent_messages):
    print(f"Message {i+1}: {message}")
